package com.ksyun.campus.client.util;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.ksyun.campus.client.EFileSystem;
import com.ksyun.campus.client.FSInputStream;
import com.ksyun.campus.client.FSOutputStream;
import com.ksyun.campus.client.domain.*;
import com.ksyun.campus.client.domain.json.DataServer;
import com.ksyun.campus.client.domain.json.MetaServer;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.BeanUtils;
import org.springframework.http.*;
import org.springframework.web.client.RestTemplate;

import javax.annotation.PostConstruct;
import java.io.File;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;


@Slf4j
public class ZkUtil {

    private static CuratorFramework curatorFramework;

    private static Gson gson;

    private static RestTemplate restTemplate;

    private static String zookeeperAddr=System.getProperty("zookeeper.addr");

    static {
        postCons();
        gson=new Gson();
        restTemplate=new RestTemplate();
    }

    @PostConstruct
    public static void postCons() {
        // 初始化，与zk建立连接，注册监听路径，当配置有变化随时更新
        // 用于处理连接失败时的重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);

        if (zookeeperAddr==null){
            zookeeperAddr="127.0.0.1:2181";
        }

        // 创建一个CuratorFramework实例
        curatorFramework = CuratorFrameworkFactory.builder()
                .connectString(zookeeperAddr)
                .sessionTimeoutMs(30 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("scr")
                .build();

        // 启动CuratorFramework实例
        curatorFramework.start();
    }

    public static ClusterInfo getCluster(){
        // 获取DataServer和MetaServer的子节点
        ClusterInfo clusterInfo = null;
        List<ClusterInfo.DataServerMsg> dataServers = null;
        try {
            List<String> dsNodes = curatorFramework.getChildren().forPath("/ds");
            List<String> msNodes = curatorFramework.getChildren().forPath("/ms");

            // 创建ClusterInfo对象
            clusterInfo = new ClusterInfo();

            // 存储DataServer信息的集合
            dataServers = new ArrayList<>();

            // 获取所有DataServer节点的信息并添加到dataServers集合中
            for (String dsNode : dsNodes) {
                byte[] bytes = curatorFramework.getData().forPath("/ds/" + dsNode);
                String dsInfo = new String(bytes);
                DataServer dataServer = gson.fromJson(dsInfo, DataServer.class);

                ClusterInfo.DataServerMsg dataServerMsg = new ClusterInfo.DataServerMsg();
                dataServerMsg.setHost(dataServer.getHost());
                dataServerMsg.setPort(dataServer.getPort());
                dataServerMsg.setCapacity(dataServer.getCapacity());
                dataServerMsg.setUseCapacity(dataServer.getUseCapacity());
//                List<String> strings = curatorFramework.getChildren().forPath("/fileList/" + dsNode);
                //获取文件数
                List<String> fileList=new ArrayList<>();
                traverseZKFiles(curatorFramework,"/fileList/"+dsNode,fileList);
                int fileTotal = fileList.size();

                dataServerMsg.setFileTotal(fileTotal);
                dataServers.add(dataServerMsg);
            }

            // 获取所有MetaServer节点的信息并设置到clusterInfo对象中
            for (String msNode : msNodes) {
                byte[] bytes = curatorFramework.getData().forPath("/ms/" + msNode);
                String msInfo = new String(bytes);
                MetaServer metaServer = gson.fromJson(msInfo, MetaServer.class);

                if (msNode.equals("master")) {
                    // 创建并设置Master MetaServer信息
                    ClusterInfo.MetaServerMsg masterMetaServerMsg = new ClusterInfo.MetaServerMsg();
                    masterMetaServerMsg.setHost(metaServer.getHost());
                    masterMetaServerMsg.setPort(metaServer.getPort());
                    clusterInfo.setMasterMetaServer(masterMetaServerMsg);
                } else if (msNode.equals("slave")) {
                    // 创建并设置Slave MetaServer信息
                    ClusterInfo.MetaServerMsg slaveMetaServerMsg = new ClusterInfo.MetaServerMsg();
                    slaveMetaServerMsg.setHost(metaServer.getHost());
                    slaveMetaServerMsg.setPort(metaServer.getPort());
                    clusterInfo.setSlaveMetaServer(slaveMetaServerMsg);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 设置ClusterInfo对象的DataServer信息
        clusterInfo.setDataServer(dataServers);

        // 返回包含集群信息的ClusterInfo对象
        return clusterInfo;
    }

    public static FSInputStream open(String fileName,String path) {
        //得到元数据服务地址
        MetaServer metaServer = getMetaServer("master");
        if (metaServer==null){
            metaServer=getMetaServer("slave");
        }

        String msHost = metaServer.getHost();
        Integer msPort = metaServer.getPort();

        String msURL = "http://" + msHost + ":" + msPort + "/open?path=" + path;

        HttpHeaders headers = new HttpHeaders();
        headers.set("fileSystem", fileName);

        HttpEntity<Object> requestEntity = new HttpEntity<>(headers);

        // 发送副本创建请求
        ResponseEntity<ApiResponse> msResponse = restTemplate.exchange(
                msURL,
                HttpMethod.GET,
                requestEntity,
                ApiResponse.class
        );

        ApiResponse msResponseBody = msResponse.getBody();

        int code = msResponseBody.getCode();
        if(code==200){
            String data = (String) msResponseBody.getData();
            Type listType = new TypeToken<List<ReplicaData>>(){}.getType();
            List<ReplicaData> replicaDataList = gson.fromJson(data, listType);
            return new FSInputStream(replicaDataList);
        }
        return null;
    }

    public static boolean mkdir(String fileName,String path) {
        MetaServer metaServer = getMetaServer("master");

        // 如果没有找到Master节点
        if (metaServer == null) {
            // master挂了的话就去访问slave
            metaServer=getMetaServer("slave");
        }

        String msHost = metaServer.getHost();
        Integer msPort = metaServer.getPort();

        String msURL = "http://" + msHost + ":" + msPort + "/mkdir?path=" + path;

        HttpHeaders headers = new HttpHeaders();

        headers.set("fileSystem", fileName);

        HttpEntity<Object> requestEntity = new HttpEntity<>(headers);

        // 发送副本创建请求
        ResponseEntity<ApiResponse> msResponse = restTemplate.exchange(
                msURL,
                HttpMethod.GET,
                requestEntity,
                ApiResponse.class
        );

        ApiResponse msResponseBody = msResponse.getBody();

        int code = msResponseBody.getCode();
        if(code==200){
            return true;
        }
        return false;
    }

    public static FSOutputStream create(String fileName,String path) {

        try {
            MetaServer metaServer = getMetaServer("master");

            // 如果没有找到Master节点
            if (metaServer == null) {
                // master挂了的话就去访问slave
                metaServer=getMetaServer("slave");
            }

            String msHost = metaServer.getHost();
            Integer msPort = metaServer.getPort();

            String msURL = "http://" + msHost + ":" + msPort + "/create?path=" + path;

            HttpHeaders headers = new HttpHeaders();

            headers.set("fileSystem", fileName);

            HttpEntity<Object> requestEntity = new HttpEntity<>(headers);

            // 向ms发送副本创建请求
            ResponseEntity<ApiResponse> msResponse = restTemplate.exchange(
                    msURL,
                    HttpMethod.GET,
                    requestEntity,
                    ApiResponse.class
            );

            ApiResponse msResponseBody = msResponse.getBody();

            // 处理副本创建的响应
            int code = msResponseBody.getCode();
            log.info("Response Code: " + code);
            String message = msResponseBody.getMessage();
            log.info("Response Message: " + message);
            String data = (String) msResponseBody.getData();
            log.info("Response Data: " + data);

            if (code == 200) {
                // 成功创建副本，解析副本的StatInfo对象
                StatInfo statInfo = gson.fromJson(data, StatInfo.class);
                List<ReplicaData> replicaDataList = statInfo.getReplicaData();
                return new FSOutputStream(replicaDataList);
            } else {
                // 处理副本创建失败的情况
                throw new IllegalStateException("创建副本失败: " + message);
            }
        } catch (Exception e) {
            e.printStackTrace();
            // 处理异常情况
            return null;
        }
    }


    public static StatInfo getFileStats(String fileName,String path) {

        MetaServer metaServer = getMetaServer("master");
        if(metaServer==null){
            metaServer=getMetaServer("slave");
        }

        String msHost = metaServer.getHost();
        Integer msPort = metaServer.getPort();
        String url = "http://" + msHost + ":" + msPort + "/stats?path=" + path;

        // 创建请求头
        HttpHeaders headers = new HttpHeaders();
        headers.set("fileSystem", fileName);
        headers.setContentType(MediaType.APPLICATION_JSON);
        HttpEntity requestEntity = new HttpEntity(headers);

        // 发送副本请求
        ResponseEntity<ApiResponse> response = restTemplate.exchange(
                url,
                HttpMethod.GET,
                requestEntity,
                ApiResponse.class
        );
        ApiResponse body = response.getBody();

        // 获取副本响应中的数据部分
        String dataJson = (String) body.getData();

        // 将dataJson转换为StatInfo对象
        StatInfo statInfo = gson.fromJson(dataJson, StatInfo.class);

        return statInfo;
    }

    public static MetaServer getMetaServer(String node){
        MetaServer metaServer = null;
        // 获取ms节点的子节点列表
        try {
            List<String> msNodes = curatorFramework.getChildren().forPath("/ms");

            // 遍历ms节点列表，在Master节点上保存元数据
            for (String msNode : msNodes) {
                if (msNode.equals(node)) {
                    byte[] bytes = curatorFramework.getData().forPath("/ms/" + msNode);
                    String msInfo = new String(bytes);
                    // 解析元数据，获取Master节点的信息
                    metaServer = gson.fromJson(msInfo, MetaServer.class);
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return metaServer;
    }

    public static boolean delete(String fileName, String path) {
        MetaServer metaServer = getMetaServer("master");

        // 如果没有找到Master节点
        if (metaServer == null) {
            // master挂了的话就去访问slave
            metaServer=getMetaServer("slave");
        }

        String msHost = metaServer.getHost();
        Integer msPort = metaServer.getPort();

        String msURL = "http://" + msHost + ":" + msPort + "/delete?path=" + path;

        HttpHeaders msheaders = new HttpHeaders();

        msheaders.set("fileSystem", fileName);

        HttpEntity<Object> requestEntity = new HttpEntity<>(msheaders);

        // 向元数据服务发送删除文件请求
        ResponseEntity<ApiResponse> msResponse = restTemplate.exchange(
                msURL,
                HttpMethod.GET,
                requestEntity,
                ApiResponse.class
        );

        ApiResponse msResponseBody = msResponse.getBody();
        int mscode = msResponseBody.getCode();

        // 向dataServer服务发送删除物理磁盘上的文件请求

        if(mscode==200){
            String data=(String) msResponseBody.getData();
            Type listType = new TypeToken<List<ReplicaData>>(){}.getType();
            List<ReplicaData> replicaDataList = gson.fromJson(data, listType);
            for (ReplicaData replicaData :replicaDataList) {
                String dsPath = replicaData.getPath();
                String dsNode = replicaData.getDsNode();
                String dsURL = "http://" + dsNode + "/delete?path=" + dsPath;

                HttpHeaders dsheaders = new HttpHeaders();

                HttpEntity<Object> dsRequestEntity = new HttpEntity<>(dsheaders);

                // 向元数据服务发送删除文件请求
                ResponseEntity<ApiResponse> dsResponse = restTemplate.exchange(
                        dsURL,
                        HttpMethod.GET,
                        dsRequestEntity,
                        ApiResponse.class
                );
                String message = dsResponse.getBody().getMessage();
                log.info(message);
            }
            return true;
        }
        return false;
    }

    public static List<StatInfo> getListFileStats(String fileName, String path) {
        MetaServer metaServer = getMetaServer("master");

        // 如果没有找到Master节点
        if (metaServer == null) {
            // master挂了的话就去访问slave
            metaServer=getMetaServer("slave");
        }

        String msHost = metaServer.getHost();
        Integer msPort = metaServer.getPort();

        String msURL = "http://" + msHost + ":" + msPort + "/listdir?path=" + path;

        HttpHeaders headers = new HttpHeaders();

        headers.set("fileSystem", fileName);

        HttpEntity<Object> requestEntity = new HttpEntity<>(headers);

        // 发送副本创建请求
        ResponseEntity<ApiResponse> msResponse = restTemplate.exchange(
                msURL,
                HttpMethod.GET,
                requestEntity,
                ApiResponse.class
        );

        ApiResponse msResponseBody = msResponse.getBody();
        int code = msResponseBody.getCode();
        if (code==200){
            List<String> statsListJson =(List<String>) msResponseBody.getData();
            List<StatInfo> statInfoList=new ArrayList<>();
            for (String json :statsListJson) {
                StatInfo statInfo = gson.fromJson(json, StatInfo.class);
                statInfoList.add(statInfo);
            }
//            Type listType = new TypeToken<List<StatInfo>>(){}.getType();

            return statInfoList;
        }
        return null;
    }


    //递归遍历出所有文件
    public static void traverseZKFiles(CuratorFramework curatorFramework, String znodePath,List<String> fileList) throws Exception {
        Stat stat = curatorFramework.checkExists().forPath(znodePath);
        if (stat != null) {
            // 遍历子节点
            for (String child : curatorFramework.getChildren().forPath(znodePath)) {
                // 构建子节点路径
                // 使用ZKPaths工具类构建子节点的完整路径。
                String childPath = ZKPaths.makePath(znodePath, child);
                // 获取子节点的元数据信息
                Stat childStat = curatorFramework.checkExists().forPath(childPath);
                if (childStat != null) {
                    byte[] bytes = curatorFramework.getData().forPath(childPath);
                    String statsJson = new String(bytes);
                    StatInfo statInfo = gson.fromJson(statsJson, StatInfo.class);
                    // 判断节点类型是否为文件
                    if (statInfo != null && statInfo.getType().equals(FileType.File)) {
                        fileList.add(childPath);
                    }else {
                        // 如果子节点有子节点，递归遍历子节点的子节点
                        traverseZKFiles(curatorFramework, childPath,fileList);
                    }
                }
            }
        } else {
            log.info("ZNode does not exist.");
        }
    }
}
